/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.amit.util;

import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author amit
 */
public abstract class Concurrency<T> {

    ConcurrentLinkedQueue<T> cQueue;
    boolean stopWhenEmpty;
    int numThreads;
    private Thread[] tArr;
    private Thread opThread;
    private boolean isRunning = false;
    private AtomicBoolean stopThreads = new AtomicBoolean(false);
    private AtomicInteger doneThreads = new AtomicInteger(0);

    public void addToQueue(T toAdd) {
        cQueue.add(toAdd);
    }

    public void addAll(Collection<T> toAdd) {
        cQueue.addAll(toAdd);
    }

    public static void sleep(long sleepTime) {
        try {
            System.out.println("SLEEPING FOR " + sleepTime + " millis ");
            System.out.flush();
            Thread.sleep(sleepTime);
            System.out.println("WOKEN UP ");
        } catch (InterruptedException ex) {
            Logger.getLogger(Concurrency.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    public Concurrency(ConcurrentLinkedQueue cQueue, boolean stopWhenEmpty, int numThreads) {
        this.cQueue = cQueue;
        this.stopWhenEmpty = stopWhenEmpty;
        this.numThreads = numThreads;
        tArr = new Thread[numThreads];
    }

    public Concurrency(boolean stopWhenEmpty, int numThreads) {
        this.cQueue = new ConcurrentLinkedQueue<T>();
        this.stopWhenEmpty = stopWhenEmpty;
        this.numThreads = numThreads;
        tArr = new Thread[numThreads];
    }

    public abstract void doThis(T queueObject) throws Exception;

    public synchronized void startThreads() {
        isRunning = true;
        for (int i = 0; i < tArr.length; i++) {
            tArr[i] = new Thread(new Runnable() {

                public void run() {
                    while (true) {
                        if (stopThreads.get()) {
                            break;
                        }
                        T obj = cQueue.poll();
                        if (obj != null) {
                            try {
                                doThis(obj);
                            } catch (Exception ex) {
                                throw new RuntimeException(ex);
                            }
                        } else if (stopWhenEmpty) {
                            break;
                        } else {
                            try {
                                Thread.sleep(50);
                            } catch (InterruptedException ex) {
                                Logger.getLogger(Concurrency.class.getName()).log(Level.SEVERE, null, ex);
                            }
                        }

                    }
                    doneThreads.incrementAndGet();
                }
            });
            tArr[i].start();
        }
    }

    public synchronized void startOperations() {
        if (!isRunning) {
            startThreads();
        }
        isRunning = true;
    }

    public synchronized void startOperationsInSeperateThread() {
        if (!isRunning) {
            opThread = new Thread(new Runnable() {

                public void run() {
                    startThreads();
                }
            });
            opThread.start();
        }
        isRunning = true;
    }

    public synchronized void stopThreadsAndReturn() {
        stopThreads.set(true);
        for (int i = 0; i < tArr.length; i++) {
            try {
                tArr[i].join();
            } catch (InterruptedException ex) {
                Logger.getLogger(Concurrency.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        isRunning = false;
    }

    public synchronized boolean isDone() {
        return (doneThreads.get() == tArr.length) | !(isRunning);
    }

    public int getQueueSize() {
        if (cQueue != null) {
            return cQueue.size();
        } else {
            return 0;
        }
    }
}